import java.io.IOException;
import java.util.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

/**
 * Mapper for the Distributed Breadth Frist Search Fist, emmits the original
 * key value pair.  Then takes the distance to this key and the links out.
 * For each link, it emits <link, dist+1:> The fact that the link value pairs
 * have a null string for thier links is handled by the reducer.
 *
 * @author Ryan Hofler
 */
public class DBFSMapper extends MapReduceBase
    implements Mapper<LongWritable, Text, Text, Text> {
    
    public void map(LongWritable key,
		    Text value,
		    OutputCollector<Text, Text> output,
		    Reporter reporter)
	throws IOException{

        String line = value.toString();
        String[] keyVal = line.split("\\t");

        String Key = keyVal[0];
        String sDist = keyVal[1];
        String[] links = null;
        if(keyVal.length > 2){
            links = keyVal[2].split(";");
            int Dist = Integer.parseInt(sDist);

            if(Dist != Integer.MAX_VALUE)
                Dist++;

            for(int x = 0; x < links.length; x++){
                if(links[x] != "")
                    output.collect(new Text(links[x]), new Text(Dist+"\t"));
            }

            output.collect(new Text(Key), new Text(sDist + "\t" + keyVal[2]));
        }
    }
}
